/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package aggr

import Aggregators._
import scala.collection.mutable.{Map => MMap}
import scala.collection.JavaConversions._

import com.google.protobuf.Descriptors._
import com.google.protobuf.Descriptors.FieldDescriptor.Type
import com.google.protobuf.Message
import com.google.protobuf.DynamicMessage

import util.JavaTypes._
import SerializedTypes._
import scala.collection.mutable.PriorityQueue

/**
 * "quantile" aggregator.
 */
class Quantile(N: Long) extends Aggregator {
    require(N > 0)

    def compare(lhs: Any, rhs: Any): Int = {
      (lhs, rhs) match {
        case (l: JDouble, r: JDouble) => l.doubleValue.compare(r.doubleValue)
        case (l: JFloat,  r: JFloat)  => l.floatValue.compare(r.floatValue)
        case (l: JInt,    r: JInt)    => l.intValue.compare(r.intValue)
        case (l: JLong,   r: JLong)   => l.longValue.compare(r.longValue)
        case (l: Message, r: Message) => 
          def f(l: List[Any], r: List[Any]): Int = (l, r) match {
            case (l::ls, r::rs) => 
              compare(l, r) match {
                case 0 => f(ls, rs)
                case x => x
              }
            case (Nil, Nil) => 0
            case _ => throw new IllegalArgumentException()
          }
          f(l.getAllFields.values.toList, r.getAllFields.values.toList)
      }
    }

    override def aggregate(data: TraversableOnce[(Any, Any)])(writer: (Any, Any) => Unit): Unit = {
      val values = data.toList.map({ _._2 }).sortWith((x, y) => compare(x, y) < 0)
      val length = values.length

      (0l to (N-1)).map { i => 
        writer(i, values.get((i.asInstanceOf[Double]/(N-1) * (length - 1).asInstanceOf[Double]).asInstanceOf[Int]) )
      }
    }

    override def getKeyType(): Option[SerializedType] = Some(Int64Type)
}

import Util._

class QuantileFactory extends AggregatorFactory {
    // 条件:
    // value typeは数値型のみ
    // もしくは、それらを組み合わせたMessage型、
    // ただし、repeatedは利用不可。optional存在しない場合には無視する。

    override def createNew(
                  keyType: SerializedType,
                  valType: SerializedType,
                  args: List[Long]): Aggregator = {
      new Quantile(args(0))
    }

    override def check(keyType: SerializedType,
              valType: SerializedType,
              args: List[SerializedType], 
              options: Map[String, SerializedType]): Option[List[String]]= {
      val aggrName = "quantile"

      val argsErrors = checkNumOfArgs(aggrName, args.length, 1) match {
        case Nil if (args(0) != Int64Type) => List("First arguments must be int type")
        case x => x
      }

      val optionErrors = checkOptionName(aggrName, options, List()) 
      val keyErrors = if (keyType != EmptyType) List("Key must be empty.") else Nil
      val valErrors = if (!isNumber(valType)) List("Contents type must be number.") else Nil

      (argsErrors ++ optionErrors ++ keyErrors ++ valErrors) match {
        case Nil => None
        case x   => Some(x)
      }
    }
}
